package com.we.risk.registerchanneldeprecated.basicinfo.dim;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;

public class TableUserPhoneInfo {
    public static final String KFKTABLE = "fdm.hhl01_user_phone_info";
    // 1673971200 dt = 2023-01-18
    public static final String QUERYSQL =
            "select uid, "
                    + "idfa as id_combine, updated_at "
                    + "from "
                    + KFKTABLE
                    + " where deleted_at is null"
                    + " and updated_at >= 1673971200"
                    + " and uid <> 0"
                    + " and idfa is not null"
                    + " and idfa <> ''"
                    + " and idfa <> '00000000-0000-0000-0000-000000000000'";

    public long uid;
    public String id_combine;
    public int updated_at;

    public TableUserPhoneInfo() {}

    public TableUserPhoneInfo(long uid, String id_combine, int updated_at) {
        this.uid = uid;
        this.id_combine = id_combine;
        this.updated_at = updated_at;
    }

    public static TableUserPhoneInfo init(long uid) {
        return new TableUserPhoneInfo(uid, "", -1);
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        tableEnv.useCatalog("cdh6_hive");
        Table tableRes = tableEnv.sqlQuery(TableUserPhoneInfo.QUERYSQL);
        DataStream<Tuple2<Boolean, TableUserPhoneInfo>> inputStream =
                tableEnv.toRetractStream(tableRes, TableUserPhoneInfo.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableUserPhoneInfo>>() {
                                    @Override
                                    public boolean filter(Tuple2<Boolean, TableUserPhoneInfo> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TableUserPhoneInfo-filter")
                        .keyBy(
                                new KeySelector<Tuple2<Boolean, TableUserPhoneInfo>, Long>() {
                                    @Override
                                    public Long getKey(Tuple2<Boolean, TableUserPhoneInfo> value)
                                            throws Exception {
                                        return value.f1.getUid();
                                    }
                                })
                        .flatMap(new userPhoneInfoRichFlatMapFunc());

        return out;
    }

    public static class userPhoneInfoRichFlatMapFunc
            extends RichFlatMapFunction<Tuple2<Boolean, TableUserPhoneInfo>, String> {
        MapState<String, Integer> phoneinfoMState;

        @Override
        public void open(Configuration parameters) throws Exception {
            phoneinfoMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<String, Integer>(
                                            "phoneinfoMState", String.class, Integer.class));
        }

        @Override
        public void flatMap(Tuple2<Boolean, TableUserPhoneInfo> value, Collector<String> out)
                throws Exception {
            long uid = value.f1.getUid();
            String id_combine = value.f1.getId_combine();

            if (!phoneinfoMState.contains(id_combine)) {
                phoneinfoMState.put(id_combine, 1);

                JSONObject outJson = new JSONObject();
                outJson.put("uid", uid);
                outJson.put("id_combine", id_combine);
                out.collect(outJson.toString());
            }
        }
    }

    public long getUid() {
        return uid;
    }

    public String getId_combine() {
        return id_combine;
    }

    public int getUpdated_at() {
        return updated_at;
    }
}
